package net.fruit.mq.client.impl.rocketmq;

import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.slf4j.Logger;

import net.fruit.mq.client.Constants;
import net.fruit.mq.client.Message;
import net.fruit.mq.client.PropertyKeyConst;
import net.fruit.mq.client.SendResult;
import net.fruit.mq.client.impl.tracehook.MQClientSendMessageHookImpl;
import net.fruit.mq.client.impl.util.ClientLoggerUtil;
import net.fruit.mq.client.trace.common.MQTraceConstants;
import net.fruit.mq.client.trace.dispatch.AsyncDispatcher;
import net.fruit.mq.client.trace.dispatch.impl.AsyncArrayDispatcher;
import net.fruit.mq.client.transaction.LocalTransactionExecuter;
import net.fruit.mq.client.transaction.TransactionProducer;
import net.fruit.mq.client.transaction.TransactionStatus;

public class TransactionProducerImpl extends MQClientAbstract implements TransactionProducer {
    private final static Logger log = ClientLoggerUtil.getClientLogger();
    private final AtomicBoolean started = new AtomicBoolean(false);
    TransactionMQProducer transactionMQProducer = null;
    private Properties properties;
    private AsyncDispatcher traceDispatcher = null;

    public TransactionProducerImpl(Properties properties, TransactionCheckListener transactionCheckListener) {
        super(properties);
        this.properties = properties;
        transactionMQProducer =
                new TransactionMQProducer((String) properties.get(PropertyKeyConst.ProducerId), new ClientRPCHook(sessionCredentials));
        boolean isVipChannelEnabled = Boolean.parseBoolean(properties.getProperty(PropertyKeyConst.isVipChannelEnabled, "false"));
        transactionMQProducer.setVipChannelEnabled(isVipChannelEnabled);

        transactionMQProducer.setTransactionCheckListener(transactionCheckListener);
        // 为Producer增加消息轨迹回发模块
        try {
            Properties tempProperties = new Properties();
            tempProperties.put(MQTraceConstants.AccessKey, sessionCredentials.getAccessKey());
            tempProperties.put(MQTraceConstants.SecretKey, sessionCredentials.getSecretKey());
            tempProperties.put(MQTraceConstants.MaxMsgSize,"128000");
            tempProperties.put(MQTraceConstants.AsyncBufferSize,"2048");
            tempProperties.put(MQTraceConstants.MaxBatchNum, "10");
            tempProperties.put(MQTraceConstants.NAMESRV_ADDR,this.getNameServerAddr());
            tempProperties.put(MQTraceConstants.InstanceName,this.buildIntanceName());
            traceDispatcher=new AsyncArrayDispatcher(tempProperties);
            traceDispatcher.start(null,this.transactionMQProducer.getInstanceName());
            this.transactionMQProducer.getDefaultMQProducerImpl().registerSendMessageHook(
                    new MQClientSendMessageHookImpl(traceDispatcher));
        } catch (Throwable e) {
            log.error("system mqtrace hook init failed ,maybe can't send msg trace data");
        }
    }

    @Override
    public void start() {
        if (started.compareAndSet(false, true)) {
            if (transactionMQProducer.getTransactionCheckListener() == null) {
                throw new IllegalArgumentException("TransactionCheckListener is null");
            }
            // TODO 完善寻址功能。要实现TransactionMQProducer.sendMessageInTransaction
            transactionMQProducer.setNamesrvAddr(this.nameServerAddr);
            try {
                transactionMQProducer.start();
            } catch (MQClientException e) {
                throw new RuntimeException(e);
            }
        }
        if(this.traceDispatcher!=null){
            this.traceDispatcher.registerShutDownHook();
        }
    }

    @Override
    public void shutdown() {
        if (started.compareAndSet(true, false)) {
            transactionMQProducer.shutdown();
        }
    }

    @Override
    public SendResult send(final Message message, final LocalTransactionExecuter executer, Object arg) {
        this.checkMQProducerServiceState(this.transactionMQProducer.getDefaultMQProducerImpl());
        org.apache.rocketmq.common.message.Message msgRMQ = RMQUtil.msgConvert(message);
        MessageAccessor.putProperty(msgRMQ, PropertyKeyConst.ProducerId, (String) properties.get(PropertyKeyConst.ProducerId));
        org.apache.rocketmq.client.producer.TransactionSendResult sendResultRMQ = null;
        try {
            sendResultRMQ = transactionMQProducer.sendMessageInTransaction(msgRMQ,
                    new org.apache.rocketmq.client.producer.LocalTransactionExecuter() {
                        @Override
                        public LocalTransactionState executeLocalTransactionBranch(org.apache.rocketmq.common.message.Message msg,
                                                                                   Object arg) {
                            String msgId = msg.getProperty(Constants.TRANSACTION_ID);
                            message.setMsgID(msgId);
                            TransactionStatus transactionStatus = executer.execute(message, arg);
                            if (TransactionStatus.CommitTransaction == transactionStatus) {
                                return LocalTransactionState.COMMIT_MESSAGE;
                            } else if (TransactionStatus.RollbackTransaction == transactionStatus) {
                                return LocalTransactionState.ROLLBACK_MESSAGE;
                            }
                            return LocalTransactionState.UNKNOW;
                        }
                    }, arg);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        if (sendResultRMQ.getLocalTransactionState() == LocalTransactionState.ROLLBACK_MESSAGE) {
            // 本地事务显式说明了有异常，此时给应用方返回一个异常。
            throw new RuntimeException("local transaction branch failed ,so transaction rollback");
        }
        SendResult sendResult = new SendResult();
        sendResult.setMessageId(sendResultRMQ.getMsgId());
        sendResult.setTopic(sendResultRMQ.getMessageQueue().getTopic());
        // if (sendResultRMQ.getTransactionId() != null) {
        // sendResult.setMessageId(sendResultRMQ.getTransactionId());
        // sendResult.setTransactionId(sendResultRMQ.getTransactionId());
        // }
        return sendResult;
    }

    @Override
    public boolean isStarted() {
        return started.get();
    }

    @Override
    public boolean isClosed() {
        return !isStarted();
    }
}
